package org.project.d.network.handler;

import com.google.common.cache.*;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.NetUtil;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.units.qual.C;
import org.project.d.network.Connection;
import org.project.d.network.NettyConnection;
import org.project.d.network.TcpConnectionManager;
import org.project.d.network.msg.BaseMessage;
import org.project.d.network.msg.Message;
import org.project.d.network.msg.enums.Protocol;
import org.project.d.network.protocol.Packet;
import org.project.d.network.protocol.UDPPacket;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author lin
 * @version V1.0
 * @description
 * @date 2020/7/22 6:26 下午
 */
@Slf4j
@ChannelHandler.Sharable
public class ConnectionHandler extends ChannelDuplexHandler {

    private final Cache<String, Connection> udpConnection;
    private final RemovalListener<String, Connection> removalListener;
    private final HeartEvent heartEvent;
    private final ConnectionEvent connectionEvent;
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    private final long udpExpireDuration;
    private final TimeUnit udpExpireUnit;
    private final Protocol protocol;

    public ConnectionHandler(ConnectionEvent connectionEvent) {
        this.connectionEvent=connectionEvent;
        this.removalListener = connectionEvent.getRemovalListener();
        this.heartEvent = connectionEvent.getHeartEvent();
        this.udpConnection = CacheBuilder.newBuilder().expireAfterAccess(connectionEvent.getUdpExpireDuration(), connectionEvent.getUdpExpireUnit()).removalListener(connectionEvent.getRemovalListener()).build();
        this.udpExpireDuration = connectionEvent.getUdpExpireDuration();
        this.udpExpireUnit = connectionEvent.getUdpExpireUnit();
        this.protocol = connectionEvent.getProtocol();
        if (!protocol.equals(Protocol.TCP) && connectionEvent.isUseBuiltInTimingThreadPool()) {
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                try {
                    udpConnection.cleanUp();
                } catch (Exception e) {
                    log.error("udpConection.cleanUp err", e);
                }
            }, 0, udpExpireDuration, udpExpireUnit);
        }
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        super.write(ctx, msg, promise);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        if (checkIsTcpChannel(ctx)) {
//            tcp
            String s = NetUtil.toSocketAddressString((InetSocketAddress) ctx.channel().remoteAddress());
            TcpConnectionManager.putIfAbsent(s, new NettyConnection());
//            if (connection != null && connection.getChannel() != null) {
//                removalListener.onRemoval(RemovalNotification.create(s, connection, RemovalCause.REPLACED));
//            }
        }
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (checkIsTcpChannel(ctx)) {
            String key = NetUtil.toSocketAddressString((InetSocketAddress) ctx.channel().remoteAddress());
            Connection remove = TcpConnectionManager.remove(key);
            removalListener(key, remove);
        }
        super.channelInactive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object obj) throws Exception {
        if (obj instanceof Packet) {
            try {
//         缓存Connection,并且把初始化放在channelActive中进行，对UDP和TCP分别进行区分
                Packet msg = ((Packet) obj);
//                Connection nettyConnection = new NettyConnection();
                SocketAddress inetSocketAddress;
                Connection connection;
                byte type = Connection.CHANNEL_TYPE_TCP;
                if (msg instanceof UDPPacket) {
                    inetSocketAddress = ((UDPPacket) msg).getAddress();
                    String s = NetUtil.toSocketAddressString((InetSocketAddress) inetSocketAddress);
                    connection = udpConnection.get(s, NettyConnection::new);
                    type = Connection.CHANNEL_TYPE_UDP;
                    connection.init(ctx.channel(), inetSocketAddress, type, msg.getSerializationType());
                } else {
                    inetSocketAddress = ctx.channel().remoteAddress();
                    connection = TcpConnectionManager.get(NetUtil.toSocketAddressString((InetSocketAddress) ctx.channel().remoteAddress()));
                    if (Objects.isNull(connection) && ctx.channel().isActive()) {
                        throw new RuntimeException("this connection not fount in tcpConection,if channelInactive this is null and this channel is active");
                    }
                    if (!connection.isConnected() && ctx.channel().isActive()) {
                        connection.init(ctx.channel(), inetSocketAddress, type, msg.getSerializationType());
                    }
                }
                Message baseMessage = new BaseMessage(msg, connection);
                super.channelRead(ctx, baseMessage);
            } finally {
                ReferenceCountUtil.release(obj);
            }
        } else {
            super.channelRead(ctx, obj);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
//            tcp 心跳处理
            log.debug("触发心跳");
            udpConnection.cleanUp();
            if (checkIsTcpChannel(ctx) && Objects.nonNull(heartEvent)) {
                String s = NetUtil.toSocketAddressString((InetSocketAddress) ctx.channel().remoteAddress());
                Connection connection = TcpConnectionManager.get(s);
                if (connection==null){
                    return;
                }
                IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
                if (connectionEvent.isHeartEventSync()){
                    execHeartEvent(connection, idleStateEvent);
                }else {
                    scheduledExecutorService.submit(()->execHeartEvent(connection,idleStateEvent));
                }
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    private void execHeartEvent(Connection connection, IdleStateEvent idleStateEvent) {
        if (idleStateEvent.state() == IdleState.WRITER_IDLE) {
            heartEvent.writerIdle(idleStateEvent, connection);
            return;
        }
        if (idleStateEvent.state() == IdleState.READER_IDLE) {
            heartEvent.readerIdle(idleStateEvent, connection);
            return;
        }
        if (idleStateEvent.state() == IdleState.ALL_IDLE) {
            heartEvent.allIdle(idleStateEvent, connection);
            return;
        }
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        if (checkIsTcpChannel(ctx)) {
//            tcp
            String s = NetUtil.toSocketAddressString((InetSocketAddress) ctx.channel().remoteAddress());
            Connection connection = TcpConnectionManager.putIfAbsent(s, new NettyConnection());
            if (connection != null) {
                removalListener(s, connection);
            }
            TcpConnectionManager.remove(s);
        }
        super.close(ctx, promise);
    }

    private boolean checkIsTcpChannel(ChannelHandlerContext ctx) {
        return Objects.nonNull(ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (checkIsTcpChannel(ctx)) {
            String key = NetUtil.toSocketAddressString((InetSocketAddress) ctx.channel().remoteAddress());
            Connection remove = TcpConnectionManager.remove(key);
            removalListener(key, remove);
        }
        super.exceptionCaught(ctx, cause);
    }

    private void removalListener(String key, Connection remove) {
        if (connectionEvent.isRemovalListenerSync()){
            removalListener.onRemoval(RemovalNotification.create(key, remove, RemovalCause.EXPLICIT));
        }else {
            scheduledExecutorService.submit(()->removalListener.onRemoval(RemovalNotification.create(key, remove, RemovalCause.EXPLICIT)));
        }
    }
}
